package com.jackrain.nea.clould.mq;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.jackrain.nea.config.PropertiesConf;
import com.jackrain.nea.util.ApplicationContextHandle;
import com.jackrain.nea.util.InstanceUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.cglib.proxy.Enhancer;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Lucky dog on 2017/4/11.
 */
@Component
public class MQMessageStartupRunner implements CommandLineRunner {

    @Autowired
    private PropertiesConf pconf;

    private Map<Subscription,MessageListener> map = new HashMap<Subscription,MessageListener>();

    @Override
    public void run(String... args) throws Exception {
        //mq是否启动
        Boolean  producer = Boolean.valueOf(pconf.getProperty("ons.mq.producer.start"));
        Boolean  consumer = Boolean.valueOf(pconf.getProperty("ons.mq.consumer.start"));
        if(producer){
            ProducerBean producerBean = (ProducerBean) ApplicationContextHandle.getBean(ProducerBean.class);
            producerBean.start();
        }
        if(consumer){
            initSubscription();
            ConsumerBean consumerBean = (ConsumerBean) ApplicationContextHandle.getBean(ConsumerBean.class);
            consumerBean.setSubscriptionTable(map);
            consumerBean.start();
        }
    }

    /**
     * 初始化订阅关系
     */
    private void initSubscription() {
        String dsPrefixs = pconf.getProperty("ons.mq.index");
        for (String dsPrefix : dsPrefixs.split(",")) {// 多个数据源
            String listenerC=pconf.getProperty("ons.mq.listenerC"+dsPrefix);
            Class classz = InstanceUtil.getClass(listenerC);
            Enhancer enhancer = new Enhancer();
            enhancer.setSuperclass(classz);
            enhancer.setCallback(new MQMessageLogIntercept());
            MessageListener messageListener = (MessageListener) enhancer.create();
            String topic=pconf.getProperty("ons.mq.topic"+dsPrefix);
            String expr=pconf.getProperty("ons.mq.expr"+dsPrefix,"*");
            Subscription subscription = new Subscription();
            subscription.setTopic(topic);
            subscription.setExpression(expr);
            map.put(subscription,messageListener);
        }
    }





}
